import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.operators.JoinOperator;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

public class JoinwithProjection {


        public static void main(String[] args) throws Exception{

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).append("/dataset_api/Java").toString();
        DataSet<Tuple3<Integer,Byte,String>> input1 =env.readCsvFile("file:///"+rootPath+"/"+"byte.csv").types(Integer.class,Integer.class, String.class).map(new MapFunction<Tuple3<Integer, Integer, String>, Tuple3<Integer, Byte, String>>() {
            @Override
            public Tuple3<Integer, Byte, String> map(Tuple3<Integer, Integer, String> integerIntegerStringTuple3) throws Exception
            {
                return Tuple3.of(integerIntegerStringTuple3.f0,(byte)integerIntegerStringTuple3.f1.intValue(),integerIntegerStringTuple3.f2);


            }
        });

        System.out.println("------------------------------输出input1------------------------------------------");
        input1.print();


    DataSet<Tuple2<Integer, Double>> input2 = env.readCsvFile("file:///"+rootPath+"/"+"tupledata.csv").types(Integer.class,Double.class);

//    DataSet<Tuple4<Integer, String, Double, Byte>>

            DataSet<Tuple4<Integer, String, Double, Byte>>
                    result =
                    input1.join(input2)
                            // key definition on first DataSet using a field position key
                            .where(0)
                            // key definition of second DataSet using a field position key
                            .equalTo(0)
                            // select and reorder fields of matching tuples
                            .projectFirst(0,2).projectSecond(1).projectFirst(1);


    System.out.println("------------------------------输出result------------------------------------------");

            result.print();


}


}
